/**
 * File: BucketNodeid2Fileposition.java
 * Created by: mhaimel
 * Created on: May 4, 2010
 * CVS:  $Id: BucketNodeid2Fileposition.java 1.0 May 4, 2010 1:11:35 PM mhaimel Exp $
 */
package uk.ac.ebi.curtain.processor.graph;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentMap;

import org.apache.commons.lang.NotImplementedException;
import org.apache.commons.lang.StringUtils;

import uk.ac.ebi.curtain.processor.AbstractQueueProcessor;
import uk.ac.ebi.curtain.util.FileIndexUtil;
import uk.ac.ebi.curtain.util.FileIndexWriter;
import uk.ac.ebi.curtain.utils.CurtainUncheckedException;
import uk.ac.ebi.curtain.utils.io.impl.FileIO;
import uk.ac.ebi.velvet.model.Supercontig;

/**
 * @author mhaimel
 *
 */
public class BucketContigid2Fileposition extends AbstractQueueProcessor<Serializable, Serializable> {
	
	private volatile ConcurrentMap<Long, Long> id2pos = null;
	
	public BucketContigid2Fileposition() {
		/* nothing*/
	}
	
	public void setId2pos(ConcurrentMap<Long, Long> id2pos) {
		this.id2pos = id2pos;
	}

	@Override
	protected Serializable buildReturnObject() {
		return null;
	}

	@Override
	public void process() {
		int cnt = 0;
		try {
			while(true){
				File dir = (File) takeInput();
				if(null != dir){
					processDir(dir);
					submitOutput(dir);
				} else if(getInputQueue().isShutdown()){
					getLog().debug("Finished queue!");
					break;// finish processing
				}				
			}			
		} catch (InterruptedException e) {
			getLog().warn("Interrupted!!",e);
			Thread.currentThread().interrupt();
		}
		getLog().debug("Finished processing items: " + cnt);
	}

	private void processDir(File dir) {
		getLog().debug("Process node ids in bucket " + dir);
		File contigIdFile = getContigIdFile(dir);
		File summaryFile = getSummaryFile(dir);
		if(!contigIdFile.exists()){
			String msg = "Expected file does not exist: ";
			getLog().error(msg + contigIdFile);
			throw new CurtainUncheckedException(msg);
		}
		Collection<Long> ctgIdColl = loadIds(contigIdFile);
		translateIds2Positions(summaryFile,ctgIdColl);
		getLog().debug("Finished processing node ids in " + dir);
	}

	private void translateIds2Positions(File summaryFile,Collection<Long> ctgIds) {
		FileIndexWriter writer = null;
		try {
			if(summaryFile.exists()){
				if(!summaryFile.delete()){
					throw new CurtainUncheckedException("Not able to delete File " + summaryFile);
				}
			}
			writer = FileIndexUtil.buildWriter(summaryFile);
			for(Long cId : ctgIds){
				Long gPos = this.id2pos.get(cId);
				Integer id = this.getContext().getGlobalFileIndex().getId(gPos);
				Long localPost = this.getContext().getGlobalFileIndex().translate2LocalPosition(gPos);
				writer.addContigPosition(id, localPost);
			}
		} finally{
			FileIO.closeQuietly(writer);
		}
	}

	private File getSummaryFile(File dir) {
		return Supercontig.getSummaryOutputFile(dir);
	}

	private Collection<Long> loadIds(File contigIdFile) {
		List<Long> contigIds = new LinkedList<Long>();
		BufferedReader in = null;
		try {
			in = new FileIO(contigIdFile).getBufferedReader();
			String line = null;
			while((line = in.readLine()) != null){
				line = line.trim();
				if(StringUtils.isNotBlank(line)){
					contigIds.add(Long.valueOf(line));
				}
			}
		} catch (IOException e) {
			throw new CurtainUncheckedException("Problems reading file " + contigIdFile,e);
		}finally{
			FileIO.closeQuietly(in);
		}
		return contigIds;
	}

	private File getContigIdFile(File dir) {
		return Supercontig.getContigIdOutputFile(dir);
	}

	@Override
	protected Serializable doProcess(Serializable input) {
		throw new NotImplementedException();
	}
	
}
